//package net.reactor;
//
//import cn.hutool.core.date.DateUtil;
//import cn.hutool.core.date.TimeInterval;
//import reactor.core.publisher.Flux;
//import reactor.core.publisher.Mono;
//import reactor.core.scheduler.Schedulers;
//
//import java.util.Arrays;
//import java.util.List;
//import java.util.concurrent.locks.Lock;
//import java.util.concurrent.locks.ReentrantLock;
//import java.util.function.Supplier;
//
//public class ParallelFluxTests {
//
//    public static void main(String[] args) {
//        ParallelFluxTests parallelFluxTests = new ParallelFluxTests();
//        //parallelFluxTests.testParallelV1();
//        parallelFluxTests.testParallelV3();
//        //parallelFluxTests.testParallelV3();
//        //parallelFluxTests.test();
//    }
//
//
//    public void test() {
//        TimeInterval timer = DateUtil.timer();
//        Flux<String> flux = Flux.fromIterable(getIds())
//                .map(id -> {
//                    String name = getName(id);
//                    int money = getMoney(id);
//                    return name + "有" + money + "元";
//                });
//
//        System.out.println("生成Flux耗时：" + timer.intervalSecond() + "秒");
//        flux.subscribe();
//         System.out.println("总耗时：" + timer.intervalSecond() + "秒");
//    }
//
//    public void testParallelV1() {
//        TimeInterval timer = DateUtil.timer();
//        Flux<String> flux = Flux.fromIterable(getIds())
//                .parallel(10)
//                .runOn(Schedulers.boundedElastic())
//                .flatMap(x -> Mono.just(getName(x)).zipWith(Mono.just(getMoney(x)), (a, b) -> a + "有" + b + "元"))
////                .flatMap(x->{
////                    Mono<String> mono = Mono.fromFuture(CompletableFuture.supplyAsync(() -> getName(x)));
////                    Mono<Integer> mono2 = Mono.fromFuture(CompletableFuture.supplyAsync(() -> getMoney(x)));
////                   return mono.zipWith(mono2, (a, b) -> a + "有" + b + "元").subscribeOn(Schedulers.parallel());
////                })
////                .log()
////                .map(id -> {
////                    String name = getName(id);
////                    int money = getMoney(id);
////                    return name + "有" + money + "元";
////                })
//                .sequential();
//
//         System.out.println("生成Flux耗时：" + timer.intervalMs() + "毫秒");
////        flux.subscribe(log::info);
////        flux.blockLast();
//        List<String> block = flux.collectList().block();
//         System.out.println(String.valueOf(block));
//         System.out.println("总耗时：" + timer.intervalMs() + "毫秒");
//    }
//
//    public void testParallelV2() {
//        TimeInterval timer = DateUtil.timer();
//
//        Flux<Integer> flux = Flux.fromIterable(getIds());
//        Flux<String> nameFlux = flux.share()
//                .parallel(5)
//                .runOn(Schedulers.boundedElastic())
//                .map(this::getName)
//                .sequential();
//        Flux<Integer> moneyFlux = flux.share()
//                .parallel(5)
//                .runOn(Schedulers.boundedElastic())
//                .map(this::getMoney)
//                .sequential();
//        Flux<String> resultFlux = Flux.zip(nameFlux, moneyFlux)
//                .map(t -> {
//                    String name = t.getT1();
//                    int money = t.getT2();
//                    return name + "有" + money + "元";
//                });
//
//         System.out.println("生成Flux耗时：" + timer.intervalMs() + "毫秒");
//        List<String> block = resultFlux.collectList().block();
//         System.out.println(String.valueOf(block));
//         System.out.println("总耗时：" + timer.intervalMs() + "毫秒");
//    }
//
//
//    class IdsSupplier implements Supplier<List<Integer>> {
//        private List<Integer> ids;
//        private final Lock lock = new ReentrantLock();
//
//        @Override
//        public List<Integer> get() {
//            lock.lock();
//            try {
//                if (ids == null) {
//                    ids = getIds();
//                }
//
//                return ids;
//            } finally {
//                lock.unlock();
//            }
//        }
//    }
//
//    public void testParallelV3() {
//        TimeInterval timer = DateUtil.timer();
//
//        IdsSupplier idsSupplier = new IdsSupplier();
//        Flux<Integer> flux = Flux.push(sink -> {
//            List<Integer> ids = idsSupplier.get();
//            ids.forEach(sink::next);
//            sink.complete();
//        });
//
//        Flux<String> nameFlux = flux.share()
//                .parallel(5)
//                .runOn(Schedulers.boundedElastic())
//                .map(this::getName)
//                .sequential();
//        Flux<Integer> moneyFlux = flux.share()
//                .parallel(5)
//                .runOn(Schedulers.boundedElastic())
//                .map(this::getMoney)
//                .sequential();
//        Flux<String> resultFlux = Flux.zip(nameFlux, moneyFlux)
//                .map(t -> {
//                    String name = t.getT1();
//                    int money = t.getT2();
//                    return name + "有" + money + "元";
//                });
//
//         System.out.println("生成Flux耗时：" + timer.intervalMs() + "毫秒");
//        List<String> block = resultFlux.collectList().block();
//         System.out.println(String.valueOf(block));
//         System.out.println("总耗时：" + timer.intervalMs() + "毫秒");
//    }
//
//    private List<Integer> getIds() {
//        try {
//            Thread.sleep(5000);
//            return Arrays.asList(1, 2, 3, 4, 5);
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
//    }
//
//    private String getName(int id) {
//        try {
//            Thread.sleep(1000);
//            return "姓名-" + id;
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
//    }
//
//    private int getMoney(int id) {
//        try {
//            Thread.sleep(1000);
//            return id * 100;
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
//    }
//}
